package com.snail.starter.kafka.producer;

import com.snail.starter.utils.json.JsonUtils;
import com.snail.starter.utils.lambda.GenericEntityBuilder;
import com.snail.starter.mq.dto.LocalMessageDTO;
import com.snail.starter.mq.dto.MQMessage;
import com.snail.starter.mq.enums.LocalMessageTypeEnum;
import com.snail.starter.mq.producer.AbstractMessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author: snail framework
 * @Date: 2023/7/4 13:43
 * @Version: 1.0.0
 */
@Slf4j
@Component
@ConditionalOnMissingBean(KafkaProperties.class)
public class KafkaProducer extends AbstractMessageProducer {

    @Resource
    KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public LocalMessageTypeEnum messageType() {
        return LocalMessageTypeEnum.KAFKA;
    }

    @Override
    public boolean send(LocalMessageDTO localMessageDTO) {
        try {
            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(getTopic(localMessageDTO.getTopic()), localMessageDTO.getBizId(),
                    JsonUtils.toJson(GenericEntityBuilder.of(MQMessage::new)
                            .with(MQMessage::setMessageId, localMessageDTO.getMessageId())
                            .with(MQMessage::setContent, localMessageDTO.getContent())
                            .with(MQMessage::setTenantId, localMessageDTO.getTenantId())
                            .build()));
            listenableFuture.get(2L, TimeUnit.SECONDS);
            return true;
        } catch (InterruptedException e) {
            log.error("", e);
            log.error("InterruptedException localMessageDTO:{}", localMessageDTO);
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            log.error("", e);
            log.error("ExecutionException localMessageDTO:{}", localMessageDTO);
        } catch (TimeoutException e) {
            log.error("", e);
            log.error("TimeoutException localMessageDTO:{}", localMessageDTO);
        } catch (Throwable e) {
            log.error("", e);
            log.error("Throwable localMessageDTO:{}", localMessageDTO);
        }
        return false;
    }
}